#include "catalog.h"
#include "query.h"
#include "sort.h"
#include "index.h"

//
// SMJ: Sort merge join
// 

int Operators::matchRec(const Record &outerRec, 
				const Record &innerRec, 
				const AttrDesc &attrDesc1, 
				const AttrDesc &attrDesc2)
{
	//compare inner and outer record

	//int
	if(attrDesc1.attrType == 0 && attrDesc2.attrType == 0)
	{
		int left, right;
		memcpy(&left, (char*)outerRec.data + attrDesc1.attrOffset, attrDesc1.attrLen);
		memcpy(&right, (char*)innerRec.data + attrDesc2.attrOffset, attrDesc2.attrLen);

		if(left > right)
			return 1;
		else if(right > left)
			return -1;
		else
			return 0;
	}
	//double
	else if(attrDesc1.attrType == 1 && attrDesc2.attrType == 1)
	{
		double left, right;
		memcpy(&left, (char*)outerRec.data + attrDesc1.attrOffset, attrDesc1.attrLen);
		memcpy(&right, (char*)innerRec.data + attrDesc2.attrOffset, attrDesc2.attrLen);

		if(left > right)
			return 1;
		else if(right > left)
			return -1;
		else
			return 0;

	}
	//char*
	else if(attrDesc1.attrType == 2 && attrDesc2.attrType == 2)
	{
		char* left;
		char* right;
		strncpy(left, (char*)outerRec.data + attrDesc1.attrOffset, attrDesc1.attrLen);
		strncpy(right, (char*)innerRec.data + attrDesc2.attrOffset, attrDesc2.attrLen);
		int cmp = strncmp(left, right, attrDesc1.attrLen);
		if(cmp > 0)
			return 1;
		else if(cmp < 0)
			return -1;
		else
			return 0;
	}
	else
	{
		cout << "Non matching attr types" << endl;
		return -2;
	}

}
Status Operators::SMJ(const string & result,          // output relation name
                      const int projCnt,              // number of attributes in the projection
                      const AttrDesc attrDescArray[], // The projection list (as AttrDesc)
                      const AttrDesc & attrDesc1,     // The left attribute in the join predicate
                      const Operator op,              // The join operation
                      const AttrDesc & attrDesc2,     // The left attribute in the join predicate
                      const int reclen)               // The lenght of a tuple in the result relation
{
  cout << "Algorithm: SM Join" << endl;
    Status lStatus;

    int attr1Cnt, attr2Cnt;
    AttrDesc *Attr1Descs;
    AttrDesc *Attr2Descs;
    lStatus = attrCat->getRelInfo(attrDesc1.relName, attr1Cnt, Attr1Descs);
    if(lStatus != OK)
        return lStatus;
    lStatus = attrCat->getRelInfo(attrDesc2.relName, attr2Cnt, Attr2Descs);
    if(lStatus != OK)
        return lStatus;

    int sf1len = 0;
    int sf2len = 0;
    for(int i = 0; i < attr1Cnt; i++)
    {
        sf1len += Attr1Descs[i].attrLen;
    }
    for(int i = 0; i < attr2Cnt; i++)
    {
        sf2len += Attr2Descs[i].attrLen;
    }

	unsigned numBufPoolPages = (unsigned)((float)bufMgr->numUnpinnedPages()* .8);//K
	int maxitems1 = numBufPoolPages * (PAGESIZE / sf1len);	
    int maxitems2 = numBufPoolPages * (PAGESIZE / sf2len);

	SortedFile sf1(attrDesc1.relName, attrDesc1.attrOffset, attrDesc1.attrLen, (Datatype)attrDesc1.attrType, maxitems1, lStatus);
	if(lStatus != OK)
		return lStatus;
	SortedFile sf2(attrDesc2.relName, attrDesc2.attrOffset, attrDesc2.attrLen, (Datatype)attrDesc2.attrType, maxitems2, lStatus);
	if(lStatus != OK)
		return lStatus;	

	HeapFile resultFile(result, lStatus);
	if(lStatus != OK)
		return lStatus;

	Record leftRec, rightRec, result_rec;

	result_rec.data = malloc(reclen);
    result_rec.length = reclen;
	bool marked = false;
	lStatus = sf1.next(leftRec);
	if(lStatus != OK)
		return lStatus;
	lStatus = sf2.next(rightRec);
	if(lStatus != OK)
		return lStatus;

	RID result_RID;

	while(true)
	{

		//if left == right
		int cmp = matchRec(leftRec, rightRec, attrDesc1, attrDesc2);	
		if(cmp == 0)
		{
            	int result_offset = 0;
            //copy recs into the result rec
		for(int i = 0; i < projCnt; i++)
		{
		    	//check all recs
                	if(strcmp(attrDescArray[i].relName, attrDesc1.relName) == 0)  
                	{
                        if(result_offset + attrDescArray[i].attrLen > result_rec.length)
                            return ATTRTOOLONG;

		        	memcpy((char*)result_rec.data + result_offset, 
                        (char*)leftRec.data + attrDescArray[i].attrOffset, attrDescArray[i].attrLen);
                    		result_offset += attrDescArray[i].attrLen;
                	}	  
                	else if(strcmp(attrDescArray[i].relName, attrDesc2.relName) == 0)  
                	{
                        if(result_offset + attrDescArray[i].attrLen > result_rec.length)
                            return ATTRTOOLONG;
                        
			        memcpy((char*)result_rec.data + result_offset,
                        (char*)rightRec.data + attrDescArray[i].attrOffset, attrDescArray[i].attrLen);
                    		result_offset += attrDescArray[i].attrLen;
                	}
		        else if(i == (projCnt - 1))
			{
				return ATTRNOTFOUND;
			}
            	}
			//copy recs into the result rec
		lStatus = resultFile.insertRecord(result_rec, result_RID);
		if(lStatus != OK)
			return lStatus;
			
		if(!marked)
		{
			lStatus = sf2.setMark();
			if(lStatus != OK)
				return lStatus;
			marked = true;
		}
		lStatus = sf2.next(rightRec);
		if(lStatus != OK && lStatus != FILEEOF)
			return lStatus;
		if(lStatus == FILEEOF)
		{
			sf2.gotoMark();
			lStatus = sf1.next(leftRec);
			if(lStatus == FILEEOF)
			{
				break;
			}
			if(lStatus != OK)
				return lStatus;
		}

	}
		//if left < right
	else if(cmp == -1)
	{
		if(marked)
        {
           	lStatus = sf2.gotoMark();
           	if(lStatus != OK)
           		return lStatus;
           	marked = false;
            sf2.next(rightRec);
       	}	
        lStatus = sf1.next(leftRec);
	    if(lStatus == FILEEOF)
		    break;    
        if(lStatus != OK)
        	return lStatus;
	}
        //if left > right
	else if(cmp == 1)
	{
        	lStatus = sf2.next(rightRec);
		if(lStatus == FILEEOF)
			break;
        	if(lStatus != OK)
                	return lStatus;
	}
	else
	{
		return NONMATCHINGATTR;
	}
}

    free(result_rec.data);

  return OK;
}

